package flink.scene;

import flink.redis_connect.RedisConnector;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.rocketmq.flink.legacy.RocketMQConfig;
import org.apache.rocketmq.flink.legacy.RocketMQSink;
import org.apache.rocketmq.flink.legacy.RocketMQSourceFunction;
import org.apache.rocketmq.flink.legacy.common.serialization.SimpleTupleDeserializationSchema;
import org.apache.rocketmq.flink.legacy.function.SinkMapFunction;
import org.apache.rocketmq.flink.legacy.function.SourceMapFunction;

import java.util.Properties;

import static org.apache.rocketmq.flink.legacy.RocketMQConfig.CONSUMER_OFFSET_LATEST;
import static org.apache.rocketmq.flink.legacy.RocketMQConfig.DEFAULT_CONSUMER_TAG;

/**
 * Created by tzs on 2022/3/22.
 */
public class FromMQToRedis {
    /**
     * Source Config
     *
     * @return properties
     */
    private static Properties getConsumerProps() {
        Properties consumerProps = new Properties();
        consumerProps.setProperty(RocketMQConfig.NAME_SERVER_ADDR, "192.168.0.104:9876");
        consumerProps.setProperty(RocketMQConfig.CONSUMER_GROUP, "flink_consumer_test");
        consumerProps.setProperty(RocketMQConfig.CONSUMER_TOPIC, "SOURCE_TOPIC");
        consumerProps.setProperty(RocketMQConfig.CONSUMER_TAG, DEFAULT_CONSUMER_TAG);
        consumerProps.setProperty(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, CONSUMER_OFFSET_LATEST);
        return consumerProps;
    }


    /**
     * 自定义redis数据类型的映射
     */
    public static class MyRedisMapper implements RedisMapper<Tuple2<String, Integer>> {

        /**
         * 设置redis数据类型以及hash表键的名称。
         * 即将处理结果(元组Tuple2<String, Integer>)映射为redis里的hash表数据类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "flink");
        }

        /**
         * 设置键，从元组中获取到单词作为键
         * @param data 元组数据
         * @return java.lang.String 键
         */
        @Override
        public String getKeyFromData(Tuple2<String, Integer> data) {
            return data.f0;
        }

        /**
         * 设置键的值，从元组中获取到单词出现的次数设置为键的值
         * @param data 元组数据
         * @return java.lang.String 键对应的值
         */
        @Override
        public String getValueFromData(Tuple2<String, Integer> data) {
            return data.f1.toString();
        }
    }

    public static void main(String[] args) throws Exception {
        try{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            //rocketmq 消费者配置
            Properties consumerProps = getConsumerProps();
            //rocketmq 消息序列化方案
            SimpleTupleDeserializationSchema schema = new SimpleTupleDeserializationSchema();
            //接收rocketmq 消息
            DataStreamSource<Tuple2<String, String>> source = env.addSource(new RocketMQSourceFunction<>(schema, consumerProps));
            source.name("获取rocketmq消息");
            source.print();

            source.map(tuple2->(String)tuple2.getField(1)).name("取出消息体")
                    .map(new MapFunction<String, Tuple2<String, Integer>>() {
                        @Override
                        public Tuple2<String, Integer> map(String s) throws Exception {
                            return new Tuple2<>(s, 1);
                        }
                    }).name("转为带count的Tuple2对象").keyBy(0).sum(1).name("聚合计数")
                    .addSink(new RedisSink<Tuple2<String, Integer>>(new FlinkJedisPoolConfig.Builder().setHost("192.168.0.104").setPort(6379).build(), new RedisConnector.MyRedisMapper())).name("存储到redis");

            // 执行任务
            env.execute("source:rocketmq-sink:redis ");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
